package com.howso.cloud.kafka.factory;/**
 * @author sjt
 * @since 2021-08-16 10:52
 */

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.Serializer;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.util.Assert;

import java.util.Map;


@Slf4j
public class KafkaProducerFactory<K, V> extends DefaultKafkaProducerFactory {

    private Map<String, Object> configs;

    private Serializer<K> keySerializer;

    private Serializer<V> valueSerializer;

    private String transactionIdPrefix;

    private volatile CloseSafeProducer<K, V> producer;

    public Map<String, Object> getConfigs() {
        return configs;
    }

    public Serializer<K> getKeySerializer() {
        return keySerializer;
    }

    @Override
    public void setKeySerializer(Serializer keySerializer) {
        this.keySerializer = keySerializer;
        super.setKeySerializer(keySerializer);
    }

    public Serializer<V> getValueSerializer() {
        return valueSerializer;
    }

    @Override
    public void setValueSerializer(Serializer valueSerializer) {
        this.valueSerializer = valueSerializer;
        super.setValueSerializer(valueSerializer);
    }

    public CloseSafeProducer<K, V> getProducer() {
        return producer;
    }

    /**
     * Construct a factory with the provided configuration.
     *
     * @param configs the configuration.
     */
    public KafkaProducerFactory(Map configs) {
        super(configs);
        this.configs = configs;
    }

    /**
     * Construct a factory with the provided configuration and {@link Serializer}s.
     *
     * @param configs         the configuration.
     * @param keySerializer   the key {@link Serializer}.
     * @param valueSerializer the value {@link Serializer}.
     */
    public KafkaProducerFactory(Map configs, Serializer keySerializer, Serializer valueSerializer) {
        super(configs, keySerializer, valueSerializer);
        this.configs = configs;
        this.keySerializer = keySerializer;
        this.valueSerializer = valueSerializer;
    }

    /**
     * 子类必须返回一个原始生产者，该生产者将被包装在{@link CloseSafeProducer}.
     *
     * @return the producer.
     */
    @Override
    public Producer<K, V> createProducer() {
        Producer producer = super.createProducer();
        this.producer = (CloseSafeProducer<K, V>) producer;
        return this.producer;
    }

    protected Producer<K, V> createKafkaProducer() {
        return new KafkaProducer<K, V>(this.configs, this.keySerializer, this.valueSerializer);
    }

    @Override
    public void destroy() {
        super.destroy();
    }

}
